package com.cloudwise.isop.drools.utils;

import com.cloudwise.isop.drools.pojo.AlertRulePO;
import com.cloudwise.isop.drools.pojo.CalculateAbridgeAlertDataDTO;
import com.cloudwise.isop.drools.pojo.SourceDataDTO;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;

import javax.annotation.concurrent.ThreadSafe;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author jeffse
 */
@ThreadSafe
public class AlertDataUtils {
    private static int DEFAULT_PENDING_MESSAGES = 5_000;
    private static int DEFAULT_SEND_TIMEOUT = 10;
    private static Map<Long, AlertRulePO> alertRulePOMap = new HashMap<>();
    private static PulsarClient pulsarClient;

    static {
        try {
            pulsarClient = PulsarClient.builder().serviceUrl(ConfigUtils.getConfig().getString("pulsar.serviceUrl")).build();
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
    }

    public static void init(List<AlertRulePO> alertRulePOS){
        alertRulePOS.forEach(item->{
            alertRulePOMap.put(item.getId(), item);
        });
    }

    public static void send(Long droolsRuleId, SourceDataDTO sourceDataDTO, String level) throws Exception {
        AlertRulePO alertRulePO = alertRulePOMap.get(droolsRuleId);
        CalculateAbridgeAlertDataDTO calculateAbridgeAlertDataDTO = new CalculateAbridgeAlertDataDTO(level, alertRulePO.getRuleId(), sourceDataDTO.getT(), sourceDataDTO.getE(), alertRulePO.getId(), sourceDataDTO.getDataCriterionNameEn(), sourceDataDTO.getCriterionPointNameEn(), sourceDataDTO.getS());
        Producer producer = pulsarClient.newProducer(Schema.STRING)
                .maxPendingMessages(DEFAULT_PENDING_MESSAGES)
                .blockIfQueueFull(true)
                .sendTimeout(DEFAULT_SEND_TIMEOUT, TimeUnit.SECONDS)
                .topic(ConfigUtils.getConfig().getString("topicName.outPutTopic"))
                .create();
        producer.sendAsync(JacksonUtils.obj2json(calculateAbridgeAlertDataDTO));
    }

}